package com.yupi.springbootinit.kafka;


import com.yupi.springbootinit.common.ErrorCode;
import com.yupi.springbootinit.es.entity.ChatSession;
import com.yupi.springbootinit.exception.BusinessException;
import com.yupi.springbootinit.kafka.contast.KafkaTopicConstant;
import com.yupi.springbootinit.model.entity.ChatSessions;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.support.SendResult;
import org.springframework.stereotype.Service;
import org.springframework.util.concurrent.ListenableFuture;

@Service
@Slf4j
public class MessageService {

    private final KafkaTemplate<String, ChatSessions> kafkaTemplate;
    @Autowired
    public MessageService(KafkaTemplate<String, ChatSessions> kafkaTemplate) {
        this.kafkaTemplate = kafkaTemplate;
    }
    public void sendDataAddMessage(ChatSessions chatSession) {
        ListenableFuture<SendResult<String, ChatSessions>> send = kafkaTemplate.send(KafkaTopicConstant.TOPIC_ADD, chatSession);
        send.addCallback(success -> {
            log.info("消息发送成功-{}",chatSession);
                }, fail -> {
                    throw new BusinessException(ErrorCode.MESSAGE_SEND_ERROR);
                }
        );
    };
}

